package com.peaksport.framework.extend.kafka;

import java.util.Optional;

import javax.transaction.Transactional;
import javax.transaction.Transactional.TxType;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;

import com.peaksport.framework.base.dao.IBaseDAO;
import com.peaksport.framework.base.service.CoreBaseService;
import com.peaksport.framework.util.PKJSONUtils;
import com.peaksport.framework.vo.JSONKafkaMsg;
import com.peaksport.pkms.base.entity.KFConsumeMsgEntity;
import com.peaksport.pkms.base.venum.KFMsgTypeEnum;

import lombok.extern.slf4j.Slf4j;

@Slf4j
public abstract class PKKafkaReceiver extends CoreBaseService {
	
	
	@KafkaListener(groupId = "groupId", topics = "topics")
	private void listener(ConsumerRecord<?, ?> record, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) throws Exception {
	    //判断是否NULL
        Optional<?> kafkaMessage = Optional.ofNullable(record.value());  
        if (kafkaMessage.isPresent()) {
        	Object message = kafkaMessage.get();
        	if (message instanceof JSONKafkaMsg) {
        		JSONKafkaMsg jsonKFMsg = (JSONKafkaMsg)message;
        		//记录消费消息
        		KFConsumeMsgEntity kfMsgEntity = recordConsumeMsg(jsonKFMsg);
        		//消费消息
        		try {
        			doConsumeMsg(jsonKFMsg);
        			kfMsgEntity.setEndTime(getCurrentTime());
        			kfMsgEntity.setConsumed(true);
        		} catch (Exception exc) {
        			log.error("消息消费异常：", exc);
        			kfMsgEntity.setEndTime(getCurrentTime());
        			throw exc;
        		} finally {
        			save(kfMsgEntity);
        		}	
        	}
        }
	}
	
	public abstract String getGroupId();
	public abstract String getTopic();
	
	public abstract void doConsumeMsg(JSONKafkaMsg jsonKFMsg) throws Exception;
	
	@Override
	public IBaseDAO<?> getDAO() {
		return null;
	}
	
	@Transactional(value=TxType.REQUIRES_NEW)
	private KFConsumeMsgEntity recordConsumeMsg(JSONKafkaMsg jsonKFMsg) {
		KFConsumeMsgEntity kfmsgEntity = new KFConsumeMsgEntity();
		kfmsgEntity.setTopic(jsonKFMsg.getTopic());
		kfmsgEntity.setMsgId(jsonKFMsg.getUniqueId());
		kfmsgEntity.setGroupId(getGroupId());
		kfmsgEntity.setType(KFMsgTypeEnum.JSON);
		kfmsgEntity.setStartTime(getCurrentTime());
		kfmsgEntity.setMsgBody(PKJSONUtils.toJSON(jsonKFMsg));
		return (KFConsumeMsgEntity) save(kfmsgEntity);
	}
	
}
